package com.ptppay.cf.starter;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.util.HashMap;
import java.util.Map;

import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;

import org.apache.log4j.Logger;
import org.web3j.protocol.Web3j;
import org.web3j.protocol.core.methods.response.EthBlock.Block;
import org.web3j.protocol.core.methods.response.EthTransaction;
import org.web3j.protocol.core.methods.response.Transaction;

import com.ptppay.cf.LaborChain;
import com.ptppay.cf.common.CommonEnums;
import com.ptppay.cf.core.business.service.CFService;
import com.ptppay.cf.domain.CoinPrice;
import com.ptppay.cf.processor.mq.MqPublisher;

import cn.finder.httpcommons.utils.JsonUtils;
import cn.finder.mq.router.DefaultTopicRouterMqttClient;
import cn.finder.wae.common.comm.AppApplicationContextUtil;
import cn.finder.wae.helper.EmailHelper;
import rx.Subscription;

/***
 * Labor Chain
 * 
 * @author finder
 *
 * 
 *
 */


public class LaborChainListenerServlet extends HttpServlet {

	private static Logger logger = Logger.getLogger(LaborChainListenerServlet.class);

	/**
	 * 
	 */
	private static final long serialVersionUID = -7958000580578197415L;

	
	Subscription block_sub;
	Subscription pendingTransaction_sub;
	Subscription pendingTransactionHash_sub;
	Subscription transaction_sub;
	
	private int LEAST_CONFIRMS=1; // 1 confirms
	
	CFService cFService;
	DefaultTopicRouterMqttClient mqttClient;
	
	
	@Override
	public void init(ServletConfig servletConfig) throws ServletException {

		String enabled = servletConfig.getInitParameter("enabled");

		if (!"true".equalsIgnoreCase(enabled)) {
			
			return;
		}
		logger.info("LaborChainListenerServlet.init  starting....  ");
		Web3j web3 = LaborChain.getWeb3jClient();

		cFService =AppApplicationContextUtil.getContext().getBean("cFService", CFService.class);
		
		mqttClient =AppApplicationContextUtil.getContext().getBean("finderMqttClient", DefaultTopicRouterMqttClient.class);
		
		
		/***
		 * 接受所有新的交易并把它们添加到区块链中
		 */
		 transaction_sub = web3.transactionObservable().subscribe(tx -> {
			// System.out.println(new Date());
			 logger.info("====transactionObservable: tx:"+tx.getHash());
			 

			 dealTx(tx);
			 
		});
		
		
		/**
	     * Create an Observable that emits newly created blocks on the blockchain.
	     *
	     * @param fullTransactionObjects if true, provides transactions embedded in blocks, otherwise
	     *                              transaction hashes
	     * @return Observable that emits all new blocks as they are added to the blockchain
	     * false参数指定我们只需要区块，而不是嵌入式交易
	     */
		 block_sub = web3.blockObservable(false).subscribe(ethBlock -> {
			
			 logger.info("==========blockObservable");
			 Block block= ethBlock.getBlock();
			 logger.info(String.format(" Height:%s", block.getNumber().intValue()+""));
			 
		});

		
		 /**
		     * Create an Observable to emit pending transactions, i.e. those transactions that have been
		     * submitted by a node, but don't yet form part of a block (haven't been mined yet).
		     *  一个新的交易被某个节点提交； 还没有被打包
		     *
		     * @return Observable to emit pending transaction hashes.
		     */
		 pendingTransactionHash_sub = web3.ethPendingTransactionHashObservable().subscribe(txHash -> {
			 logger.info("====ethPendingTransactionHashObservable: txHash:"+txHash);
			 
			 EthTransaction ethTx=null;
			 Transaction tx=null;
			 try{
				 
				 ethTx= LaborChain.getTransactionInfo(txHash);
				 
				 tx= ethTx.getResult();
				 if(tx==null){
					 return ;
				 }
				 
				 dealTx(tx);

			 }
			catch(Throwable e){
				 logger.error(e);
				 new EmailHelper().sendEmailToAdminInThread(String.format(" ethPendingTransactionHashObservable txHash: [%s] 异常:", txHash), e.toString());
			}
			 finally{
				 
				 
			 }
		});
		 
		/***
		 * 接受所有已经提交到网络中的等待处理的交易。(它们必须分在同一个区块中。)
		 *  
		 */
		 /**
		     * Create an Observable to emit all pending transactions that have yet to be placed into a
		     * block on the blockchain.
		     * 监控进行中的交易  这些交易还没写进区块
		     * @return Observable to emit pending transactions
		     */
		 pendingTransaction_sub = web3.pendingTransactionObservable().subscribe(tx -> {
			 
			// logger.info("====pendingTransactionObservable: tx:"+tx.getBlockHash());
			
		});
		
		
		 

		

		
	}
	
	
	/***
	 * 同意处理交易
	 * @param tx
	 */
	private void dealTx(Transaction tx){
		
		String txHash=tx.getHash();
		 try{
			 
			 // 买币取现
			 String to=tx.getTo(); //平台账户
			 
			 //BigInteger height= tx.getBlockNumber(); // block number
			 
			// BigInteger value=tx.getValue(); // coin amt
			 
			 //find  address <=> atm,if exists, do cash out ,else ignore . 判断  to 是否有绑定  to cache expire 2min
			 
			 Map<String,Object> bindItem= cFService.findBindingByWalletAddressByMchAddress(to);
			 if(bindItem!=null){
				 
				 logger.info("===the tx is now binding equipment");
				 
				 //计算金额 
				 
				BigInteger tx_fee= tx.getGas().multiply(tx.getGasPrice());//0.946
				
				String eqt_id=bindItem.get("eqt_id").toString();
				
				//compare to pretx
				Map<String,Object> pretx=cFService.findLatestingPreTxInfo(eqt_id, to, CommonEnums.PRETXS_BUY_SELL_SELL.getValue());
				
				if(pretx!=null){
					
					
					 
					 logger.info("=== exist latest transaction binding merchant wallet address is "+to);
					//tx is exists
					double pretx_money_amt=Double.parseDouble(pretx.get("money_amt").toString());
					double need_coin_amt=Double.parseDouble(pretx.get("coin_amt").toString());
					//update chain tx to pretx
					
					String pretx_id=pretx.get("id").toString();
					
					
					
					double value=tx.getValue().doubleValue()/1000000000000000000.0;
					
					String eqt_code=cFService.findEquipmentCode(eqt_id);
					
					
					 new EmailHelper().sendEmailToAdminInThread(String.format(" merchant wallet(%s)  received coin ", to),
							 String.format("eqt_code:%s, pre_txid:%s ,money_amt:%s,coin_amt:%s ", eqt_code,pretx_id,pretx_money_amt,value));
						
					
					
					cFService.updatePreTxByChainTx(pretx_id, tx.getHash(), value+"");
					
					//save chain txinfo
					
					// saveTxInfo(tx);
					
					// send out cash command
					
					
					
					
					//ETH:RMB=1:5000
					//double exchange_rate=Double.parseDouble(cFService.findEqtParam(eqt_code, "lab").get("valuestr").toString());
					
					CoinPrice coinPrice=cFService.findCoinPrice(eqt_code, CommonEnums.COIN_TYPE_LAB.getValue());
					double exchange_rate=coinPrice.getBuy_price();// receive coin price
					double miner_fee=Double.parseDouble(coinPrice.getMiner_fee().toString());
					//出钞金额 (元)
					
					/*Map<String,Object> returnMap=new HashMap<String,Object>();
					returnMap.put("eqt_code", eqt_code);
					returnMap.put("pre_tx_id", pretx_id);
					returnMap.put("money_amt", money);
					
					returnMap.put("status", "C");
					returnMap.put("msg", "outcash.normal");*/
					// 设备里面缺钱 应该也需要通知设备
					
					
					if(value<need_coin_amt){
						// act_coin amt letter  need coin amt
						
						//转的太多
						Double money_amt=value*exchange_rate;
						//去整数
						int money=money_amt.intValue();
						Map<String,Object> returnData=new HashMap<String,Object>();
						returnData.put("eqt_code", eqt_code);
						returnData.put("pre_tx_id", pretx_id);
						returnData.put("money_amt", pretx_money_amt);
						returnData.put("money_type", pretx.get("money_type"));
						returnData.put("status", "L");
						returnData.put("msg", "outcash.transfer.coin.is.less");
						sendToClient(eqt_code,returnData);
					}else if(value>=need_coin_amt){
						
						BigDecimal bg = new BigDecimal(value-need_coin_amt);  
			            double diff_coin = bg.setScale(8, BigDecimal.ROUND_HALF_UP).doubleValue();
						
						if(diff_coin>miner_fee){
							//转的太多  多与的打印交易
							Double money_amt=need_coin_amt*exchange_rate;
							//去整数
							int money=money_amt.intValue();
							
							Map<String,Object> returnData=new HashMap<String,Object>();
							returnData.put("eqt_code", eqt_code);
							returnData.put("pre_tx_id", pretx_id);
							returnData.put("money_amt", pretx_money_amt);
							returnData.put("money_type", pretx.get("money_type"));
							returnData.put("status", "B");
							returnData.put("msg", "outcash.transfer.coin.is.more");
							returnData.put("more_money_amt", (value-need_coin_amt));
							// 设备里面缺钱 应该也需要通知设备
							sendToClient(eqt_code,returnData);
						}else{
							
							//出钞金额 (元)
							
							Double money_amt=need_coin_amt*exchange_rate;
							//去整数
							int money=money_amt.intValue();
							//是否还需要判断 准备出钞金额 和 提交金额
							
							
							Map<String,Object> returnMap=new HashMap<String,Object>();
							returnMap.put("eqt_code", eqt_code);
							returnMap.put("pre_tx_id", pretx_id);
							returnMap.put("money_amt", pretx_money_amt);
							returnMap.put("money_type", pretx.get("money_type"));
							returnMap.put("status", "C");
							returnMap.put("msg", "outcash.normal");
							// 设备里面缺钱 应该也需要通知设备
							sendToClient(eqt_code,returnMap);
						}
						
						
					}
					
					
					
					
					
					
					//sendToClient(eqt_code,returnMap);
					/*String returnData=JsonUtils.getJsonString4JavaPOJO(returnMap);
					String topic=MqPublisher.TOPIC_CF_OUTCASH;
					try {
						//mqttClient.publishMsgOne(topic, returnData, eqt_code);
						MqPublisher.publish(topic, returnMap, eqt_code);
						
					} catch (Throwable e) {
						
						//记录到日志 ，并且发邮件
						new EmailHelper().sendEmailToAdminInThread(String.format(" MQ publish exception .topic: %s,payload:%,to_client_id:%s, exception detail:%s", topic,returnData,eqt_code,e.toString()), e.toString());
						
					} */
					
				}
				
				
				
			 }
			 
			 
		 }
		catch(Throwable e){
			 logger.error(e);
			 new EmailHelper().sendEmailToAdminInThread(String.format(" ethPendingTransactionHashObservable txHash: [%s] 异常:", txHash), e.toString());
		}
		 finally{
			 
			 
		 }
	}

	@Override
	public void destroy() {
		block_sub.unsubscribe();
		pendingTransaction_sub.unsubscribe();
		pendingTransactionHash_sub.unsubscribe();
		transaction_sub.unsubscribe();
	}
	
	
	/***
	 * save chain tx , not need. we can find tx detail to block info view
	 * @param trans
	 */
	private void saveTxInfo(Transaction trans){
		
		Map<String,Object> txInfo=new HashMap<>();
		txInfo.put("tx_hash", trans.getHash());
		txInfo.put("from_address", trans.getFrom());
		txInfo.put("to_address", trans.getTo());
		//txInfo.put("amt", trans.getValue());
		//txInfo.put("block_number", trans.getBlockNumber());
		txInfo.put("block_number", "");
		//txInfo.put("creates", trans.getCreates());
		txInfo.put("creates", "");
		txInfo.put("gas", trans.getGas());
		txInfo.put("gas_price", trans.getGasPrice());
		txInfo.put("gas_price_raw", trans.getGasPriceRaw());
		txInfo.put("gas_raw", trans.getGasRaw());
		txInfo.put("input", trans.getInput());
		txInfo.put("nonce", trans.getNonce());
		txInfo.put("nonce_raw", trans.getNonceRaw());
		//txInfo.put("publick_key", trans.getPublicKey());
		txInfo.put("publick_key", "");
		txInfo.put("r", trans.getR());
		//txInfo.put("raw", trans.getRaw());
		txInfo.put("raw", "");
		txInfo.put("s", trans.getS());
		txInfo.put("transaction_index", trans.getTransactionIndex());
		txInfo.put("value", trans.getValue());
		txInfo.put("value_raw", trans.getValueRaw());
		txInfo.put("token_type", CommonEnums.LABOR_COIN_TYPE.getValue());
		txInfo.put("tx_fee", trans.getGas().multiply(trans.getGasPrice()));
		/*
		txInfo.put("has_error", eth_trans.hasError()?"1":"0");
		if(eth_trans.hasError()){
			txInfo.put("error_code", eth_trans.getError().getCode());
			txInfo.put("error_data", eth_trans.getError().getData());
			txInfo.put("error_message", eth_trans.getError().getMessage());
		}*/
		txInfo.put("status", CommonEnums.TX_STATUS_COMPLETE.getValue());
		
		cFService.saveTxInfo(txInfo);
	}
	
	private void sendToClient(String eqt_code,Map<String,Object> returnMap){
		
		
		// 设备里面缺钱 应该也需要通知设备
		
		
		String returnData=JsonUtils.getJsonString4JavaPOJO(returnMap);
		String topic=MqPublisher.TOPIC_CF_OUTCASH;
		try {
			//mqttClient.publishMsgOne(topic, returnData, eqt_code);
			MqPublisher.publish(topic, returnMap, eqt_code);
			
		} catch (Throwable e) {
			
			//记录到日志 ，并且发邮件
			new EmailHelper().sendEmailToAdminInThread(String.format(" MQ publish exception .topic: %s,payload:%,to_client_id:%s, exception detail:%s", topic,returnData,eqt_code,e.toString()), e.toString());
			
		} 
	}

}
